package org.example.reactive;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {

    // 定义流中间操作处理器，既是一个发布者，也是一个订阅者
    static class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String>{
        private Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("processor订阅绑定完成");
            this.subscription = subscription;
            subscription.request(1);// 请求一个新数据
        }

        @Override
        public void onNext(String item) {
            item += "哈哈";
            // 把加工后的数据发出去
            submit(item);
            subscription.request(1);// 继续请求一个新数据
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {

        }
    }
    public static void main(String[] args) throws InterruptedException {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                System.out.println(Thread.currentThread() + "订阅开始了" + subscription);
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread() + "开始处理下一个item,item=" + item);
                if ("p-5".equals(item)) {
                    throw new IllegalArgumentException("error");
                }
                if ("p-3".equals(item)) {
                    subscription.cancel();
                }
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread() + "发生错误");
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread() + "所有数据处理完成");
            }
        };

//        publisher.subscribe(subscriber);
        MyProcessor myProcessor = new MyProcessor();
        publisher.subscribe(myProcessor);
        myProcessor.subscribe(subscriber);

        for (int i = 0; i < 10; i++) {
            publisher.submit("p-" + i);
        }

        publisher.close();

        Thread.sleep(5000);
    }
}
